package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/1 15:35
 */
public class DWD_09_DwdTradeOrderRefundApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_09_DwdTradeOrderRefundApp().init("DwdTradeOrderRefundApp", 3009, 1, "DwdTradeOrderRefundApp");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 需要关联3张表:
        // 1. 读取ods数据, 从中过滤出退单数据  order_refund_info   insert
        // 2. 筛选订单数据, 找到订单id和省份  更新字段是order_stats 并且 order_stats从1002变成1005 操作类型是update
        // 3. 字典表
        // TODO 3. 从 Kafka 读取 ods_db 数据，封装为 Flink SQL 表
        tEnv.executeSql(
            "create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string, string>, " +
                "`old` map<string, string>, " +
                "`proc_time` as PROCTIME(), " +
                "`ts` string " +
                ")" + SQLUtil.getKafkaDDL("DwdTradeOrderRefundApp", "ods_db"));
        
        // TODO 4. 读取退单表数据
        Table orderRefundInfo = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['refund_type'] refund_type, " +
                "data['refund_num'] refund_num, " +
                "data['refund_amount'] refund_amount, " +
                "data['refund_reason_type'] refund_reason_type, " +
                "data['refund_reason_txt'] refund_reason_txt, " +
                "data['create_time'] create_time, " +
                "proc_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'order_refund_info' " +
                "and `type` = 'insert' ");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);
        // TODO 5. 读取订单表数据，筛选退单数据   提出出来 订单id和省份id
        Table orderInfoRefund = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['province_id'] province_id, " +
                "`old` " +
                "from topic_db " +
                "where `table` = 'order_info' "
            //                +
            //                "and `type` = 'update' " +
            //                "and data['order_status']='1005' " +
            //                "and `old`['order_status'] is not null"
        );
        
        tEnv.createTemporaryView("order_info_refund", orderInfoRefund);
        
        // TODO 6. 建立 MySQL-LookUp 字典表
        tEnv.executeSql(SQLUtil.getBaseDicLookUpDDL());
        
        // TODO 7. 关联三张表获得退单宽表
        Table resultTable = tEnv.sqlQuery(
            "select  " +
                "ri.id, " +
                "ri.user_id, " +
                "ri.order_id, " +
                "ri.sku_id, " +
                "oi.province_id, " +
                "date_format(ri.create_time,'yyyy-MM-dd') date_id, " +
                "ri.create_time, " +
                "ri.refund_type, " +
                "type_dic.dic_name, " +
                "ri.refund_reason_type, " +
                "reason_dic.dic_name, " +
                "ri.refund_reason_txt, " +
                "ri.refund_num, " +
                "ri.refund_amount, " +
                "ri.ts, " +
                "current_row_timestamp() row_op_ts " +
                "from order_refund_info ri " +  // 退单表
                "left join  " +
                "order_info_refund oi " + // 订单表
                "on ri.order_id = oi.id " +
                "join  " +
                "base_dic for system_time as of ri.proc_time as type_dic " +
                "on ri.refund_type = type_dic.id " +
                "join " +
                "base_dic for system_time as of ri.proc_time as reason_dic " +
                "on ri.refund_reason_type=reason_dic.id");
        tEnv.createTemporaryView("result_table", resultTable);
        
        //        resultTable.execute().print();
        // TODO 8. 建立 Upsert-Kafka dwd_trade_order_refund 表
        tEnv.executeSql(
            "create table dwd_trade_order_refund( " +
                "id string, " +
                "user_id string, " +
                "order_id string, " +
                "sku_id string, " +
                "province_id string, " +
                "date_id string, " +
                "create_time string, " +
                "refund_type_code string, " +
                "refund_type_name string, " +
                "refund_reason_type_code string, " +
                "refund_reason_type_name string, " +
                "refund_reason_txt string, " +
                "refund_num string, " +
                "refund_amount string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(id) not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL("dwd_trade_order_refund"));
        
        // TODO 9. 将关联结果写入 Upsert-Kafka 表
        tEnv.executeSql("insert into dwd_trade_order_refund select * from result_table");
    }
    
}
